package com.imooc.bilibili.websocket;

import com.alibaba.fastjson.JSONObject;
import com.imooc.bilibili.constant.rocket.UserMomentsConstant;
import com.imooc.bilibili.domain.video.Danmu;
import com.imooc.bilibili.user.service.UserService;
import com.imooc.bilibili.util.StringUtils;
import com.imooc.bilibili.util.TokenUtil;
import com.imooc.bilibili.util.json.JsonUtils;
import com.imooc.bilibili.util.redis.RedisUtils;
import com.imooc.bilibili.util.rocket.RocketMQUtils;
import com.imooc.bilibili.util.spring.SpringUtils;
import com.imooc.bilibili.video.IDanmuService;
import com.imooc.bilibili.video.impl.DanmuServiceImpl;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import sun.nio.cs.US_ASCII;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 多例模式
 *
 * @author gcq
 * @Create 2022-06-07
 */
@Component
// 用来发现对应服务端口
@ServerEndpoint("/imserver/{token}")
public class WebSocketService {

    private final Logger logger = LoggerFactory.getLogger(WebSocketService.class);

    /**
     * 记录连接数量
     * 保证线程安全 原子性
     */
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);

    /**
     * 保证线程安全存储数据
     *
     * 存储链接
     */
    public static final ConcurrentHashMap<String, WebSocketService> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    /**
     * 会话
     * 客户端与服务端进行通信
     *
     */
    private Session session;

    /**
     * 连接唯一Id
     */
    private String sessionId;
    /**
     *
     */
    private Long userId;
    /**
     * 多例模式
     *
     */
    @Autowired
    private RedisUtils redisUtils;

    /**
     * 全局 applicationContext
     */
    private static ApplicationContext applicationContext;

    public static void setApplicationContext(ApplicationContext applicationContext) {
        WebSocketService.applicationContext = applicationContext;
    }

    public Session getSession() {
        return session;
    }

    public void setSession(Session session) {
        this.session = session;
    }

    /**
     * 打开链接
     *
     * @param session
     */
    @OnOpen
    public void openConnection(Session session, @PathParam("token") String token) {
        try {
            userId = TokenUtil.verifyToken(token);
        } catch (Exception e) {
            e.printStackTrace();
        }
        this.sessionId = session.getId();
        this.session = session;
        if(WEBSOCKET_MAP.contains(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            WEBSOCKET_MAP.put(sessionId, this);
        } else {
            WEBSOCKET_MAP.put(sessionId, this);
            ONLINE_COUNT.getAndIncrement();
        }
        logger.info("用户连接成功：" + sessionId + " 当前在线人数为: " + ONLINE_COUNT.get());
        try {
            this.sendMessage("0");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @OnClose
    public void closeConnection() {
        if(WEBSOCKET_MAP.contains(sessionId)) {
            WEBSOCKET_MAP.remove(sessionId);
            ONLINE_COUNT.getAndDecrement();
        }
        logger.info("用户退出: " + sessionId + "当前在线人数为: " + ONLINE_COUNT.get());
    }

    /**
     * 接受信息
     * 群发消息，效率不好
     * 数据量大的话，需要使用消息队列，缓存，不能直接循环
     * 性能优化建议：
     * 接收到用户发送弹幕后，将弹幕推送到当前所有在线的客户端，同事将弹幕保存到redis和db中
     *
     */
    @OnMessage
    public void onMessage(String message) {
        logger.info("用户信息: " + sessionId + "，报文 " + message);
        if(StringUtils.isNotEmpty(message)){
            try {
                // 群发消息
                for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()) {
                    WebSocketService webSocketService = entry.getValue();

                    // 异步发送消息到MQ中，交给MQ进行处理
                    DefaultMQProducer danmusProducer = SpringUtils.getBean("danmusProducer");
                    JSONObject jsonObject = new JSONObject();
                    jsonObject.put("sessionId", this.sessionId);
                    jsonObject.put("message", message);
                    Message msg = new Message(UserMomentsConstant.TOPIC_DANMUS, jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
                    RocketMQUtils.asyncSendMsg(danmusProducer, msg);
                    logger.info("异步发送消息成功，消息Id:{}", this.sessionId);
                }
                if(this.userId != null) {
                    // 保存弹幕到数据库中
                    Danmu danmu = JsonUtils.jsonToPojo(message, Danmu.class);
                    danmu.setCreateTime(new Date());
                    danmu.setUserId(userId);
                    IDanmuService danmuService = (IDanmuService) applicationContext.getBean("danmuService");
                    // TODO 也可以放到消息队列红中，队列进行消费
                    danmuService.addDanmus(danmu);
                    // 保存弹幕至redis，异步或者同步都可以，redis本身比较快
                    // 分布式情况下需要分布式锁，可以放到队列中，异步保存redis数据
                    danmuService.addDanmusToRedis(danmu);
                    logger.info("消息同步至redis和db成功");
                }
            } catch (Exception e) {
                logger.error("弹幕接收出现问题: {}", e);
                e.printStackTrace();
            }
        }
    }

    @OnError
    public void onError(Throwable error) {

    }

    /**
     * 发送信息给到客户端
     *
     * @param message
     * @throws IOException
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 每隔5000毫秒【每隔5秒】执行一次该方法
     * 获取当前在线人数,循环遍历发送到多个客户端中
     * TODO 优化建议：异步发送
     *
     */
    @Scheduled(fixedRate = 5000)
    public void noticeOnLineCount0() throws IOException {
        for (Map.Entry<String, WebSocketService> entry : WEBSOCKET_MAP.entrySet()) {
            WebSocketService webSocketService = entry.getValue();
            if(webSocketService.getSession().isOpen()){
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("onLineCount", ONLINE_COUNT.get());
                jsonObject.put("msg", "当前在线人数为" + ONLINE_COUNT.get());
                webSocketService.sendMessage(jsonObject.toJSONString());
            }
        }
    }
}